package mq

import (
    "context"
    "encoding/json"
    "fcs-api/src/app/model"
    "fcs-api/src/app/service/invoke"
    "fcs-api/src/client/rabbitmq"
    "fmt"
    "time"

    "git.inspur.com/x/common/loggerfactory"
    cloudevents "github.com/cloudevents/sdk-go/v2"
    "github.com/streadway/amqp"
)

var logger = loggerfactory.GetLogger()
var mq rabbitmq.MqClient

//ListenOssMq listen oss trigger mq
func ListenOssMq(mqurl string, ch chan int) {
    var err error
    for {
        mq, err = rabbitmq.InitClient(mqurl)
        if err != nil {
            _ = logger.Error("Init oss mq client error: ", err)
        } else {
            break
        }
        logger.Info("reconnect oss mq...")
        time.Sleep(1 * time.Minute)
    }
    logger.Info("Receiving oss mq message ...")

    //listen mq status and reconnect
    go func(mqurl string, ch chan int) {
        reason, ok := <-mq.NotifyClose(make(chan *amqp.Error))
        // exit this goroutine if closed by developer
        if !ok {
            logger.Info("connection closed")
        }
        logger.Info("oss mq connection closed, reason:", reason)
        logger.Info("重新初始化oss mq")
        close(ch)
        stopCh := make(chan int)
        go ListenOssMq(mqurl, stopCh)
    }(mqurl, ch)

    //receive oss mq
    err = mq.Receive(model.OssMqExchange, model.OssMqQueue, model.OssMqRk, func(msg amqp.Delivery) {
        logger.Info("Receive oss event message: ", string(msg.Body))

        m := map[string][]map[string]interface{}{}
        err := json.Unmarshal(msg.Body, &m)
        if err != nil {
            logger.Info("Unmarshal msg.Body err: ", err)
            return //todo or break
        }
        ruleId := m["Records"][0]["s3"].(map[string]interface{})["configurationId"]
        logger.Info("ruleId: ", ruleId)
        t, err := model.GetTriggerByRuleId(ruleId.(string))
        if err != nil {
            logger.Info("GetTriggerByRuleId err: ", err)
            return
        }
        logger.Debug("get trigger by ruleId: ", t)
        //调用函数
        c, err := cloudevents.NewDefaultClient()
        if err != nil {
            _ = logger.Error("CallFunction: failed to create client, %v", err)
            return
        }

        url := fmt.Sprintf("%s/sync/%s/%s/%s.%s", model.Base.FunctionInvokeHost, t.A5tId, t.GroupName, t.FuncName, t.TargetName)
        event := cloudevents.NewEvent(cloudevents.VersionV1)
        event.SetType(model.OssEventType)
        event.SetSource(model.OssEventSource)
        if err := event.SetData(cloudevents.ApplicationJSON, invoke.Messages(string(msg.Body))); err != nil {
            _ = logger.Error("CallFunction: failed set data %v", err)
            return
        }

        ctx := cloudevents.ContextWithTarget(context.Background(), url)
        // 发送cloud event请求并获取结果
        _, result := c.Request(ctx, event)
        if !cloudevents.IsACK(result) {
            _ = logger.Error("failed to send oss event: %v", result)
            return
        }
    })
    if err != nil {
        _ = logger.Error("Receive oss mq error: ", err)
    }

    select {
    case <-ch:
        logger.Info("stop bad oss mq")
        return
    }
}
